package bjut.zrpc.core.optimization.netty;

import bjut.zrpc.core.optimization.bean.ZRpcServiceBean;
import bjut.zrpc.core.optimization.constant.ZRpcConstant;
import bjut.zrpc.core.optimization.enums.ZRpcCompressEnum;
import bjut.zrpc.core.optimization.enums.ZRpcConfigEnum;
import bjut.zrpc.core.optimization.enums.ZRpcResponseEnum;
import bjut.zrpc.core.optimization.enums.ZRpcSerializerEnum;
import bjut.zrpc.core.optimization.factory.SingletonFactory;
import bjut.zrpc.core.optimization.handler.ZRpcRequestHandler;
import bjut.zrpc.core.optimization.interfaces.Interceptor;
import bjut.zrpc.core.optimization.message.ZRpcMessage;
import bjut.zrpc.core.optimization.message.ZRpcRequestMessage;
import bjut.zrpc.core.optimization.message.ZRpcResponseMessage;
import bjut.zrpc.core.optimization.protect.LimitStream;
import bjut.zrpc.core.optimization.utils.PropertiesReadUtil;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.SocketAddress;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;


/**
 * @Author: zhuzhiming
 * @Date: 2022/6/24 11:56
 */
@Component
public class NettyServerHandler extends SimpleChannelInboundHandler {

    private static final ZRpcRequestHandler requestHandler = SingletonFactory.getInstance(ZRpcRequestHandler.class);
    private final LimitStream limitStream;
    private final String beforeMethod = "beforeHandlerInterceptor";
    private final String afterMethod = "afterHandlerInterceptor";
    private  ZRpcServiceBean zRpcServiceBean;


    public NettyServerHandler() {
        limitStream = SingletonFactory.getInstance(LimitStream.class);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {


        try {
            if (msg instanceof ZRpcMessage) {

                ZRpcMessage zRpcMessage = (ZRpcMessage) msg;
                byte messageType = zRpcMessage.getMessageType();
                String compress = PropertiesReadUtil.getConfig(ZRpcConfigEnum.ZRPC_COMPRESS.getPropertyValue()[0]);
                String serializer = PropertiesReadUtil.getConfig(ZRpcConfigEnum.ZRPC_SERIALIZER.getPropertyValue()[0]);
                zRpcMessage.setCodec(serializer.equals("true") ? ZRpcSerializerEnum.HESSIAN.getCodec() : ZRpcSerializerEnum.getSerializationType(serializer));
                zRpcMessage.setCompress(compress.equals("true") ? ZRpcCompressEnum.GZIP.getType() : ZRpcCompressEnum.getCompressType(compress));

                ZRpcRequestMessage zRpcRequestMessage = (ZRpcRequestMessage) zRpcMessage.getBodyData();
                String serviceInfo = zRpcRequestMessage.getServiceInfo();

                if (messageType == ZRpcConstant.HEART_REQUEST_TYPE) {
                    zRpcMessage.setMessageType(ZRpcConstant.HEART_RESPONSE_TYPE);
                } else {
                    this.zRpcServiceBean = limitStream.getServiceBean(serviceInfo);
                    zRpcMessage.setMessageType(ZRpcConstant.RESPONSE_TYPE);

                    if (zRpcServiceBean.isNeedInterceptor()) {
                        zRpcRequestMessage = doBeforeInterceptorMethod(zRpcRequestMessage);
                    }
                    if (zRpcServiceBean.isLimit()) {
                        boolean needLimit = checkLimit(zRpcServiceBean, ctx.channel().remoteAddress());
                        if (needLimit) {
                            ZRpcResponseMessage<Object> rpcResponseMessage = ZRpcResponseMessage.fail(ZRpcResponseEnum.FAIL, zRpcRequestMessage.getRequestId());
                            zRpcMessage.setBodyData(rpcResponseMessage);
                            ctx.writeAndFlush(zRpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                            return;
                        }
                    }

                    Object result = null;
                    if (zRpcServiceBean.isConcurrencyControl()) {
                        Semaphore semaphore = zRpcServiceBean.getSemaphore();
                        try {
                            result = requestHandler.handler(zRpcRequestMessage, zRpcServiceBean, semaphore);
                            if (result == null) {
                                ZRpcResponseMessage<Object> rpcResponseMessage = ZRpcResponseMessage.fail(ZRpcResponseEnum.FAIL, zRpcRequestMessage.getRequestId());
                                zRpcMessage.setBodyData(rpcResponseMessage);
                                return;
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            semaphore.release(1);
                        }

                    } else {
                        result = requestHandler.handler(zRpcRequestMessage, null, null);
                    }

                    if (ctx.channel().isActive() && ctx.channel().isWritable()) {
                        ZRpcResponseMessage<Object> responseMessage = ZRpcResponseMessage.success(result, zRpcRequestMessage.getRequestId());
                        zRpcMessage.setBodyData(responseMessage);
                    } else {
                        ZRpcResponseMessage<Object> rpcResponseMessage = ZRpcResponseMessage.fail(ZRpcResponseEnum.FAIL, zRpcRequestMessage.getRequestId());
                        zRpcMessage.setBodyData(rpcResponseMessage);
                    }

                }

                if (zRpcServiceBean.isNeedInterceptor()) {
                    zRpcMessage = doAfterInterceptorMethod(zRpcMessage);
                }
                ctx.writeAndFlush(zRpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

            }
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage());
        } finally {
            ReferenceCountUtil.release(msg);
        }

    }

    private ZRpcMessage doAfterInterceptorMethod(ZRpcMessage zRpcMessage) {

        ZRpcMessage result;

        String interceptor = zRpcServiceBean.getInterceptor();
        try {
            Class<?> aClass = Class.forName(interceptor);
            boolean isInterceptor = Interceptor.class.isAssignableFrom(aClass);
            if (!isInterceptor) {
                throw new RuntimeException(aClass.getSimpleName() + " must implements Interceptor");
            }
            Object instance = aClass.newInstance();
            Method method = aClass.getMethod(afterMethod, ZRpcMessage.class);
            result = (ZRpcMessage) method.invoke(instance, zRpcMessage);
        } catch (ClassNotFoundException | InstantiationException | IllegalAccessException | NoSuchMethodException | InvocationTargetException e) {
            throw new RuntimeException(e.getMessage());
        }
        return result;

    }

    private ZRpcRequestMessage doBeforeInterceptorMethod(ZRpcRequestMessage zRpcRequestMessage) {

        ZRpcRequestMessage result;

        String interceptor = zRpcServiceBean.getInterceptor();
        try {
            Class<?> aClass = Class.forName(interceptor);
            boolean isInterceptor = Interceptor.class.isAssignableFrom(aClass);
            if (!isInterceptor) {
                throw new RuntimeException(aClass.getSimpleName() + " must implements Interceptor");
            }
            Object instance = aClass.newInstance();
            Method method = aClass.getMethod(beforeMethod, ZRpcRequestMessage.class);
            result = (ZRpcRequestMessage) method.invoke(instance, zRpcRequestMessage);
        } catch (ClassNotFoundException | InstantiationException | IllegalAccessException | NoSuchMethodException | InvocationTargetException e) {
            throw new RuntimeException(e.getMessage());
        }
        return result;
    }


    private boolean checkLimit(ZRpcServiceBean zRpcServiceBean, SocketAddress remoteAddress) {
        String serviceNameAndClientAddress = zRpcServiceBean.getRpcServiceName() + "-" + remoteAddress;
        return limitStream.needLimit(serviceNameAndClientAddress, zRpcServiceBean.getLimitCount(), zRpcServiceBean.getInLimitTime(), zRpcServiceBean.getLimitDuration());
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

        if (evt instanceof IdleStateEvent) {
            if (((IdleStateEvent) evt).state() == IdleState.READER_IDLE) {
                ctx.channel().close();
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        ctx.channel().close();
    }
}
